#include "svc_sync.h"
#include <message.pb.h>
namespace svc_file
{
    CacheDelete::CacheDelete(const std::shared_ptr<sw::redis::Redis>& redis, 
                            const bitemq::MQClient::ptr& mq_client,
                            const bitemq::declare_settings& declare_settings)
                            : redis_(redis),
    {
        publisher_ = std::make_shared<bitemq::publisher>(mq_client,declare_settings);
        subscriber_ = std::make_shared<bitemq::subscriber>(mq_client,declare_settings);
        subscriber_->consume(std::bind(&CacheDelete::callback, this, std::placeholders::_1, std::placeholders::_2));
    }

    void CacheDelete::sync(const std::string& cache_key)
    {
        try
        {
            auto rdx = redis_->transaction(false,false);
            auto r = rdx.redis();
            r.del(cache_key);
            api_message::DeleteCacheMsg msg;
            msg.add_key(cache_key);
            bool ret = publisher_->publish(msg.SerializeAsString());
            if(!ret)
            {
                throw vp::error::VPException(vp::error::VPError::MQ_ERROR);
            }
            catch(const vp::error::VPException& e)
            {
                ERR("CacheDelete::sync publish cache_key:{} failed, error:{}", cache_key, e.what());
                throw;
            }
            catch(const std::exception& e)
            {
                ERR("CacheDelete::sync publish cache_key:{} failed, error:{}", cache_key, e.what());
                throw vp::error::VPException(vp::error::VPError::SERVER_INTERNAL_ERROR);
            }
            rdx.exec();
        }
    }

    void CacheDelete::callback(const char* body,size_t len)
    {
        api_message::DeleteCacheMsg msg;
        bool ret = msg.ParseFromArray(body,len);
        if(!ret)
        {
            ERR("CacheDelete::callback parse message failed");
            return;
        }   

        auto rdx = redis_->transaction(false,false);
        auto r = rdx.redis();
        int sz = msg.keys_size();
        for(int i = 0; i < sz; i++)
        {
            r.del(msg.keys(i));
            DBG("CacheDelete::callback del key:{}", msg.keys(i));
        }
        rdx.exec();
    }
}